package beautiful.butterfly.drds.binlog.consumer;

import beautiful.butterfly.drds.binlog.connection.ConnectionAttributes;
import beautiful.butterfly.drds.binlog.connection.ShardingRule;
import beautiful.butterfly.drds.binlog.snapshot.SnapShot;
import beautiful.butterfly.drds.binlog.utils.Constants;
import beautiful.butterfly.drds.binlog.utils.DataUtils;
import beautiful.butterfly.drds.binlog.utils.TimeUtils;
import beautiful.butterfly.drds.binlog.utils.WriteResultUtils;
import com.alibaba.fastjson.JSONObject;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;

import java.util.ArrayList;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

@Slf4j
public class Consumer
{


    @Setter
    @Getter
    private int index;
    @Setter
    @Getter
    private BlockingQueue<SnapShot> snapShotQueue = null;
    @Setter
    @Getter
    private Thread thread = null;

    public Consumer(int index)
    {
        this.index = index;
        this.snapShotQueue = new LinkedBlockingQueue<SnapShot>(1 * 10000);// 1万条
    }


    public long getQueueLeftSize()
    {
        return snapShotQueue.remainingCapacity();
    }

    // poll: 若队列为空，返回null。
    // remove:若队列为空，抛出NoSuchElementException异常。
    // take:若队列为空，发生阻塞，等待有元素。
    public Consumer start()
    {
        // 1)
        Runnable runnable = new Runnable()
        {
            @Override
            public void run()
            {
                // 开始处理队列里的每一个数据
                SnapShot snapShot = null;
                while (true)
                {// 无限循环中,本线程永远不会退出
                    // LoggerUtils.info(logger, "snapShotQueue("+index+").size:
                    // "+snapShotQueue.size());
                    try
                    {// 尝试拿取数据
                        snapShot = snapShotQueue.take();
                    } catch (InterruptedException e)
                    {
                        snapShot = null;
                    }
                    // 为空则继续
                    if (null == snapShot)
                    {

                        TimeUtils.sleepMilliSeconds(10);// 主动休眠10ms
                        continue;
                    }
                    // 成功拿到了一个SnapShot对象
                    // 1)检测是否成功标志
                    if (false == snapShot.getGlobalValid().get())
                    {
                        snapShot.setWriteResult(WriteResultUtils.FAIL.ordinal());
                        continue;
                    }
                    // 当前有效
                    // type肯定是IO类型，否则不会加入到这里
                    int result;
                    if (snapShot.isAccepted())
                    {// 确实需要处理
                        result = DataUtils.handleMulti(snapShot.getDatabaseName(), snapShot.getTableName(), snapShot.getDatas())
                                ? WriteResultUtils.SUCCEED.ordinal() : WriteResultUtils.FAIL.ordinal();
                    } else
                    {
                        result = WriteResultUtils.SUCCEED.ordinal();
                    }
                    // 将结果写回到result
                    snapShot.setWriteResult(result);
                    // 如果写入失败，设置全局失败标志
                    if (WriteResultUtils.FAIL.ordinal() == result)
                    {
                        snapShot.getGlobalValid().set(false);
                    }
                }
            }
        };
        // 2)
        thread = new Thread(runnable);
        // 3)
        thread.setDaemon(true);// 后台线程
        thread.setName("consumer_thread_" + index);
        thread.start();
        log.info(thread.getName() + " started successfully...");
        return this;

    }

    // put---无空间会等待
    // add--- 满时,立即返回,会抛出异常
    // offer---满时,立即返回,不抛异常
    public void addSnapShot(SnapShot snapShot, ConnectionAttributes connectionAttributes)
    {
        if (null == snapShot)
        {
            return;
        }
        // 增加分库分表的改写规则,这里是最后的机会改写
        // 放在最后，也是为了防止破坏之前已经稳定的业务逻辑，避免引入新的bug
        if (snapShot.isAccepted())
        {// 如果是需要丢弃的数据，就不要浪费时间了
            ShardingRule shardingRule = connectionAttributes.findShardingRule(snapShot.getDatabaseName(), snapShot.getTableName());
            if (null != shardingRule)
            {// 确实命中了一个分库分表规则
                updateSnapShotSelf(snapShot, shardingRule.getDestinationDatabaseNamePattern(), shardingRule.getDestinationTableNamePattern());
                updateSnapShotData(shardingRule.getDestinationDatabaseNamePattern(), shardingRule.getDestinationTableNamePattern(), snapShot.getDatas());
            }
        }
        //
        // 坚决不能阻塞，否则selector会挂住
        while (true == snapShot.getGlobalValid().get())
        {
            // 如果还是有效的，则需要进入
            try
            {
                snapShotQueue.add(snapShot);
                break;
            } catch (Exception e)
            {
                continue;
            }
        }
    }

    private void updateSnapShotSelf(SnapShot snapShot, String databaseName, String tableName)
    {
        snapShot.setDatabaseName(databaseName);
        snapShot.setTableName(tableName);
    }

    //
    private void updateSnapShotData(String databaseName, String tableName, ArrayList<JSONObject> jsonObjects)
    {
        if (null == jsonObjects)
        {
            return;
        }
        // 遍历
        // {"optType":"UPDATE","databaseName":"skyeye","id":2,"optTime":1461740209000,"tableName":"aaa"}
        for (JSONObject jsonObject : jsonObjects)
        {
            jsonObject.put(Constants.database_name, databaseName);
            jsonObject.put(Constants.table_name, tableName);
        }
    }

}